package com.atguigu.flink.chapter01_wordcount;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/10/19
 *
 *      流批一体： 一套代码，既可以流计算也可以批计算。只需要进行简单的设置即可
 *            流式计算由于不可靠性，通常需要使用批计算进行对数。
 *
 */
public class Demo3_StreamingBatchOne
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //启动一个线程去算。不设置，默认用当前环境中所有的cpu作为线程数进行计算
        //env.setParallelism(1);

        /*
                AUTOMATIC: 自动根据数据源选择。
                                无界流：  选择流处理
                                有界流：  选择批处理

                BATCH: 指定代码进行批处理运算。
                            只有有界流才能运行成功。

                STREAMING: 指定是流式处理。
                                可以处理有界流。

         */
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        //使用环境读取数据源，获取一个流
        //DataStreamSource<String> source = env.socketTextStream("hadoop103", 8888);
        DataStreamSource<String> source = env.readTextFile("data/words.txt");

        //对source进行转换，单词统计
        source
            .flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>()
            {
                /*
                        对DataSource的数据进行处理
                            line： 一行数据
                            Collector<String> out： 负责收集输出的数据

                            (hello,1):  Tuple
                 */
                @Override
                public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {
                    String[] words = line.split(" ");
                    for (String word : words) {
                        out.collect(Tuple2.of(word, 1));
                    }
                }
            })
            //类似之前的groupBy, keyBy代表按照key进行分组
            .keyBy(new KeySelector<Tuple2<String, Integer>, String>()
            {
                //从输入的Tuple2<String, Integer> 返回key部分
                @Override
                public String getKey(Tuple2<String, Integer> value) throws Exception {
                    return value.f0;
                }
            })
            .sum(1)
            .print();



        //启动执行环境，计算才会开始，永不结束
        env.execute();

    }
}
